//
// Created by 郝羽 on 2018/6/19.
//

#include "metric_server.hpp"

#include <arpa/inet.h>
#include <glib.h>
#include <gio/gio.h>
#include <list>
#include <map>
#include <rapidjson/document.h>
#include <gio/gunixinputstream.h>

#include "metric.hpp"

struct MetricServer;

struct MetricConn {
    MetricServer* server;
    GSocketConnection* conn;
    char* remote_address;
    char* buf;
    int   buf_len;
    int   buf_cap;
};

struct MetricServer {
    GSocketService* service;

    guint16 port;

    MetricsByTime metrics;
};

void parse_metric(Metric& metric, const gchar* data, gsize len) {
    rapidjson::Document json;
    json.Parse(data, len);

    if (json.FindMember("time") != json.MemberEnd()) {
        metric.time = json["time"].GetInt();
    }

    if (json.FindMember("rtt") != json.MemberEnd()) {
        auto rtt = json["rtt"].GetArray();
        for (auto i = 0; i < rtt.Size(); i++) {
            metric.rtt.Update(rtt[i].GetInt());
        }
    }

    if (json.FindMember("jitter") != json.MemberEnd()) {
        auto jitter = json["jitter"].GetArray();
        for (auto i = 0; i < jitter.Size(); i++) {
            metric.jitter.Update(jitter[i].GetInt());
        }
    }

    if (json.FindMember("setup-time") != json.MemberEnd()) {
        auto setup_time = json["setup-time"].GetArray();
        for (auto i = 0; i < setup_time.Size(); i++) {
            metric.setup_time.Update(setup_time[i].GetInt());
        }
    }

    if (json.FindMember("inited") != json.MemberEnd()) {
        metric.inited.Update(json["inited"].GetInt());
    }

    if (json.FindMember("setup") != json.MemberEnd()) {
        metric.setup.Update(json["setup"].GetInt());
    }

    if (json.FindMember("send") != json.MemberEnd()) {
        metric.send.Update(json["send"].GetInt64());
    }

    if (json.FindMember("recv") != json.MemberEnd()) {
        metric.recv.Update(json["recv"].GetInt64());
    }

    if (json.FindMember("send-pkts") != json.MemberEnd()) {
        metric.send_packet.Update(json["send-pkts"].GetInt64());
    }

    if (json.FindMember("recv-pkts") != json.MemberEnd()) {
        metric.recv_packet.Update(json["recv-pkts"].GetInt64());
    }

    if (json.FindMember("ping-time") != json.MemberEnd()) {
        auto ping_time = json["ping-time"].GetArray();
        for (auto i = 0; i < ping_time.Size(); i++) {
            metric.ping_time.Update(ping_time[i].GetInt());
        }
    }
}

void metric_server_update(MetricServer *server, const Metric &metric) {
    auto it = server->metrics.find(metric.time);
    if (it == server->metrics.end()) {
        server->metrics.emplace(metric.time, metric);
    } else {
        auto& old = it->second;
        old.rtt.Update(metric.rtt);
        old.jitter.Update(metric.jitter);
        old.setup_time.Update(metric.setup_time);
        old.inited.Update(metric.inited);
        old.setup.Update(metric.setup);
        old.send.Update(metric.send);
        old.recv.Update(metric.recv);
        old.send_packet.Update(metric.send_packet);
        old.recv_packet.Update(metric.recv_packet);
    }
}

void metric_conn_close(MetricConn* conn) {
    g_message("conn '%s' close", conn->remote_address);
    if (conn->remote_address) {
        g_free(conn->remote_address);
        conn->remote_address = nullptr;
    }

    if (conn->buf && conn->buf_cap) {
        g_slice_free1(conn->buf_cap, conn->buf);
        conn->buf = nullptr;
        conn->buf_len = 0;
        conn->buf_cap = 0;
    }

    auto sock = g_socket_connection_get_socket(conn->conn);
    g_socket_close(sock, nullptr);
    g_object_unref(conn->conn);

    delete(conn);
}

gboolean metric_conn_read(GPollableInputStream *in, MetricConn* conn) {
    //MetricConn* conn = static_cast<MetricConn*>(data);
    if (conn->buf == nullptr) {
        uint32_t l = 0;
        GError* err = nullptr;
        auto n_read = g_pollable_input_stream_read_nonblocking(in, &l, 4, nullptr, &err);
        if (n_read == 0) {
            metric_conn_close(conn);
            return G_SOURCE_REMOVE;
        } else if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK)) {
            return G_SOURCE_CONTINUE;
        } else if (n_read < 0) {
            g_message("conn '%s' read header error: %s", conn->remote_address, err->message);
            metric_conn_close(conn);
            return G_SOURCE_REMOVE;
        }

        l = htonl(l);
        g_message("conn '%s' receive msg header: len=%d", conn->remote_address, l);
        conn->buf = (char*)g_slice_alloc(l);
        conn->buf_cap = l;
        conn->buf_len = 0;
    }

    GError* err = nullptr;
    auto n_read = g_pollable_input_stream_read_nonblocking(in, conn->buf, conn->buf_cap - conn->buf_len, nullptr, &err);
    if (n_read == 0) {
        metric_conn_close(conn);
        return G_SOURCE_REMOVE;
    } else if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK)) {
        return G_SOURCE_CONTINUE;
    } else if (n_read < 0) {
        g_message("conn '%s' read payload error: %s", conn->remote_address, err->message);
        metric_conn_close(conn);
        return G_SOURCE_REMOVE;
    }

    conn->buf_len += n_read;
    if (conn->buf_len == conn->buf_cap) {
        g_message("conn '%s' read complete", conn->remote_address);
        Metric m;
        parse_metric(m, conn->buf, conn->buf_len);
        metric_server_update(conn->server, m);
        g_slice_free1(conn->buf_cap, conn->buf);
        conn->buf = nullptr;
        conn->buf_len = 0;
        conn->buf_cap = 0;
    }

    return G_SOURCE_CONTINUE;
}

//gboolean metric_conn_timeout(gpointer user_data) {
//    auto conn = static_cast<MetricConn*>(user_data);
//    metric_conn_close(conn);
//}

gboolean metric_server_handle_incoming(GSocketService* service,
        GSocketConnection* connection, GObject* source, gpointer user_data) {

    auto server = static_cast<MetricServer*>(user_data);
    MetricConn* conn = new MetricConn();
    conn->server = server;
    conn->conn = (GSocketConnection*)g_object_ref(connection);
    conn->buf = nullptr;
    conn->buf_len = 0;
    conn->buf_cap = 0;

    auto sock = g_socket_connection_get_socket(connection);
    auto sockaddr = g_socket_get_remote_address(sock, nullptr);
    GInetAddress *addr = g_inet_socket_address_get_address(G_INET_SOCKET_ADDRESS(sockaddr));
    guint16 port = g_inet_socket_address_get_port(G_INET_SOCKET_ADDRESS(sockaddr));
    conn->remote_address = g_strdup_printf("%s:%d", g_inet_address_to_string(addr), port);

    auto stream = g_io_stream_get_input_stream(G_IO_STREAM(connection));
    auto in_source = g_pollable_input_stream_create_source(G_POLLABLE_INPUT_STREAM(stream), nullptr);
    g_source_set_callback(in_source, (GSourceFunc)metric_conn_read, conn, nullptr);
    g_source_attach(in_source, nullptr);

//    conn->timer_source = g_timeout_source_new(3000);
//    g_source_set_callback(conn->timer_source, metric_conn_timeout, conn, nullptr);
//    g_source_attach(conn->timer_source, nullptr);

//    gint fd = g_socket_get_fd(conn->sock);
//
//    GIOChannel* ch = g_io_channel_unix_new(fd);
//
//    g_io_add_watch(ch, G_IO_IN, metric_conn_read, conn);
//
//    g_io_channel_unref(ch);
    return G_SOURCE_CONTINUE;
}

MetricServer* metric_server_open(Context& ctx) {

    MetricServer* server = new MetricServer();

    GError* err = nullptr;

    server->service = g_socket_service_new();

    server->port = g_socket_listener_add_any_inet_port(
            G_SOCKET_LISTENER(server->service), nullptr, &err);

    if (err != nullptr) {
        g_warning("open metric server error: %s", err->message);
        exit(-1);
    }

    g_signal_connect(server->service, "incoming", G_CALLBACK(metric_server_handle_incoming), server);

    g_socket_service_start(server->service);

    return server;
}

uint16_t metric_server_port(MetricServer* server) {
    return server->port;
}

MetricsByTime metric_server_metrics(MetricServer* server) {
    return server->metrics;
}

void metric_server_close(MetricServer* server) {
    if (server->service) {
        g_socket_service_stop(server->service);
        g_object_unref(server->service);
        server->service = nullptr;
    }

    delete(server);
}